package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo17Cache {
  def main(args: Array[String]): Unit = {
    // 构建Spark上下文环境
    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo17Cache")
    conf.setMaster("local")
    val sc: SparkContext = new SparkContext(conf)

    // 设置checkpoint保存的目录
    sc.setCheckpointDir("Spark/data/stu/checkpoint")

    // 1、读取students、scores数据
    val stuRDD: RDD[String] = sc.textFile("Spark/data/stu/students.txt")


    val mapStuRDD: RDD[String] = stuRDD.map(line => {
      // 默认情况下 每个job会打印1000次 说明每个job都读取了一次文件
      println("========student========")
      line
    })

    // 对多次使用的RDD进行缓存
    // 直接调用cache方法即可将数据缓存到内存中
    //    mapStuRDD.cache()
    // 如果当内存不够的时候 需要将数据放入Dick磁盘
    // 需要使用persist() 方法 传入一个缓存策略级别
    //    mapStuRDD.persist(StorageLevel.DISK_ONLY) // 将RDD的数据缓存到磁盘

    mapStuRDD.cache()

    /**
     * checkpoint（会启动一个新的job）将RDD中的数据保存到HDFS
     * 在checkpoint之前先进行cache或者persist
     * 主要是出于容错考虑的，主要应用在SparkStreaming中
     */
    mapStuRDD.checkpoint()

    // 统计班级人数
    val clazzKVRDD: RDD[(String, Int)] = mapStuRDD
      .map(line => (line.split(",")(4), 1))

    val clazzCnt: RDD[(String, Int)] = clazzKVRDD
      .reduceByKey((i, j) => i + j)

    clazzCnt
      .foreach(println)

    // 统计性别人数
    val genderKVRDD: RDD[(String, Int)] = mapStuRDD
      .map(line => (line.split(",")(3), 1))

    val genderCnt: RDD[(String, Int)] = genderKVRDD
      .reduceByKey(_ + _)

    genderCnt
      .foreach(println)

    // 用完记得释放缓存
    stuRDD.unpersist()

    while (true) {

    }


  }

}
